package org.monkey.socket.plugin.provider;

import cn.hutool.core.util.StrUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.monkey.common.constant.entity.ErrorResultEnum;
import org.monkey.common.constant.exception.BaseException;
import org.monkey.socket.plugin.config.SocketConstant;
import org.monkey.socket.plugin.config.SocketMqConfig;
import org.monkey.socket.plugin.service.IMsgProducer;
import org.springframework.beans.factory.InitializingBean;

/**
 * 对接中间件 消息消费的接口 搭配 IMsgProducer使用，如果需要弃用默认rocketMq消息发布订阅实现,那么只需重新实现IMsgProducer和IMsgConsumer并注入
 */
@Slf4j
@Data
public class DefaultIMsgProducer implements IMsgProducer, InitializingBean {

  private DefaultMQProducer producer;
  private SocketMqConfig socketMqConfig;

  public DefaultIMsgProducer(SocketMqConfig socketMqConfig) {
    this.socketMqConfig = socketMqConfig;
  }

  @Override
  public void afterPropertiesSet() {
    if (producer == null) {
      producer = new DefaultMQProducer(
          SocketConstant.SOCKET_GROUP + "-" + socketMqConfig.getProfile());
      producer.setNamesrvAddr(socketMqConfig.getNameServerAddr());
    }
    try {
      producer.setInstanceName(StrUtil.uuid().replace("-", ""));
      producer.start();
      log.info("socket msg mq publish producer init success,nameserver={},topic={},group={}",
          socketMqConfig.getNameServerAddr(),
          SocketConstant.SOCKET_TOPIC + "-" + socketMqConfig.getProfile(),
          SocketConstant.SOCKET_GROUP + "-" + socketMqConfig.getProfile());
    } catch (MQClientException e) {
      log.error("socket msg mq publish producer init failed,nameserver={},topic={},group={},",
          socketMqConfig.getNameServerAddr(),
          SocketConstant.SOCKET_TOPIC + "-" + socketMqConfig.getProfile(),
          SocketConstant.SOCKET_GROUP + "-" + socketMqConfig.getProfile(), e);
    }
  }


  @Override
  public void producer(byte[] bytes) {
    Message message = new Message(SocketConstant.SOCKET_TOPIC + "-" + socketMqConfig.getProfile(),
        bytes);
    try {
      log.info("publish messages ....");
      producer.send(message);
    } catch (Exception e) {
      log.error("msg send failed:", e);
      throw new BaseException(ErrorResultEnum.SYSTEM_ERROR);
    }
  }

}
